/*
 *    Copyright 2014-2017 Alexey Danilov
 *    Copyright 2021 Institute of Software Chinese Academy of Sciences, ISRC

 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package com.danikula.videocache;

import com.danikula.videocache.defaultStrategy.DefaultCacheStrategy;
import com.danikula.videocache.interfacers.CacheListener;
import com.danikula.videocache.interfacers.ReSetAfterErrorListener;
import com.danikula.videocache.onefileStrategy.OneFileCacheStrategy;
import com.danikula.videocache.util.LogUtil;
import ohos.agp.utils.TextTool;
import ohos.app.Context;
import ohos.utils.net.Uri;

import com.danikula.videocache.file.DiskUsage;
import com.danikula.videocache.file.FileNameGenerator;
import com.danikula.videocache.file.Md5FileNameGenerator;
import com.danikula.videocache.file.TotalCountLruDiskUsage;
import com.danikula.videocache.file.TotalSizeLruDiskUsage;
import com.danikula.videocache.headers.EmptyHeadersInjector;
import com.danikula.videocache.headers.HeaderInjector;
import com.danikula.videocache.sourcestorage.SourceInfoStorage;
import com.danikula.videocache.sourcestorage.SourceInfoStorageFactory;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.danikula.videocache.Preconditions.checkAllNotNull;
import static com.danikula.videocache.Preconditions.checkNotNull;

/**
 * Simple lightweight proxy server with file caching support that handles HTTP requests.
 * Typical usage:
 * <pre><code>
 * public onCreate(Bundle state) {
 *      super.onCreate(state);
 *
 *      HttpProxyCacheServer proxy = getProxy();
 *      String proxyUrl = proxy.getProxyUrl(VIDEO_URL);
 *      videoView.setVideoPath(proxyUrl);
 * }
 *
 * private HttpProxyCacheServer getProxy() {
 * // should return single instance of HttpProxyCacheServer shared for whole app.
 * }
 * </code></pre>
 *
 * @author Alexey Danilov (danikula@gmail.com).
 */
public class HttpProxyCacheServer {
    private static final String PROXY_HOST = "127.0.0.1";
    private static final int ATTEMPTS_TIME = 3;
    private static final String TAG = "HttpProxyCacheServer";

    private final Object clientsLock = new Object();
    private final ExecutorService socketProcessor = Executors.newSingleThreadExecutor();
    private final Map<String, HttpProxyCacheServerClients> clientsMap = new ConcurrentHashMap<>();
    private final ServerSocket serverSocket;
    private final int port;
    private final Thread waitConnectionThread;
    private final Config config;
    private final Pinger pinger;
    private long cacheOffset = -1;
    private long sameOffsetTimes;  // 相同offset连续请求次数
    private int attemptsTime = ATTEMPTS_TIME; // 重试次数
    private ReSetAfterErrorListener listener; // 当前策略出错客户端重置操作监听器
    private String originUrl = "";

    public HttpProxyCacheServer(Context context) {
        this(new Builder(context).buildConfig());
    }

    private HttpProxyCacheServer(Config config) {
        this.config = checkNotNull(config);
        try {
            InetAddress inetAddress = InetAddress.getByName(PROXY_HOST);
            this.serverSocket = new ServerSocket(0, 8, inetAddress);
            this.port = serverSocket.getLocalPort();
            IgnoreHostProxySelector.install(PROXY_HOST, port);
            CountDownLatch startSignal = new CountDownLatch(1);
            this.waitConnectionThread = new Thread(new WaitRequestsRunnable(startSignal));
            this.waitConnectionThread.start();
            startSignal.await(); // freeze thread, wait for server starts
            this.pinger = new Pinger(PROXY_HOST, port);
            LogUtil.info(this.getClass().getSimpleName(), "Proxy cache server started. Is it alive? " + isAlive());
        } catch (IOException | InterruptedException e) {
            socketProcessor.shutdown();
            throw new IllegalStateException("Error starting local proxy server", e);
        }
    }

    public void setAttemptsTime(int attemptsTime) {
        this.attemptsTime = attemptsTime;
        if (this.attemptsTime <= ATTEMPTS_TIME) {
            this.attemptsTime = ATTEMPTS_TIME;
        }
    }

    public void setListener(ReSetAfterErrorListener listener) {
        this.listener = listener;
    }

    public Config getConfig() {
        return config;
    }

    /**
     * Returns url that wrap original url and should be used for client (MediaPlayer, ExoPlayer, etc).
     * <p>
     * If file for this url is fully cached (it means method {@link #isCached(String)} returns {@code true})
     * then file:// uri to cached file will be returned.
     * <p>
     * Calling this method has same effect as calling {@link #getProxyUrl(String, boolean)} with 2nd parameter set to
     * {@code true}.
     *
     * @param url a url to file that should be cached.
     * @return a wrapped by proxy url if file is not fully cached or url pointed to cache file otherwise.
     */
    public String getProxyUrl(String url) {
        this.originUrl = url;
        return getProxyUrl(url, true);
    }

    /**
     * Returns url that wrap original url and should be used for client (MediaPlayer, ExoPlayer, etc).
     * <p>
     * If parameter {@code allowCachedFileUri} is {@code true} and file for this url is fully cached
     * (it means method {@link #isCached(String)} returns {@code true}) then file:// uri to cached file will be
     * returned.
     *
     * @param url                a url to file that should be cached.
     * @param allowCachedFileUri {@code true} if allow to return file:// uri if url is fully cached
     * @return a wrapped by proxy url if file is not fully cached or url pointed to cache file otherwise (if {@code
     * allowCachedFileUri} is {@code true}).
     */
    public String getProxyUrl(String url, boolean allowCachedFileUri) {
        if (allowCachedFileUri && isCached(url)) {
            File cacheFile = getCacheFile(url);
            touchFileSafely(cacheFile);
            removeTempDirectory(cacheFile);
            return Uri.getUriFromFile(cacheFile).toString();
        }
        return isAlive() ? appendToProxyUrl(url) : url;
    }

    /**
     * 分片缓存缓存完毕之后 读取本地缓存的时候检测一下
     * 之前的存储碎片文件的临时文件夹如果还没有删除
     * 则把他以及他里面的碎片文件都删除
     *
     * @param cacheFile 当前临时缓存文件
     */
    public void removeTempDirectory(File cacheFile) {
        try {
            if (config.cacheStrategy instanceof DefaultCacheStrategy ||
                    config.cacheStrategy instanceof OneFileCacheStrategy) {
                return;
            }
            config.diskUsage.removeTemp(cacheFile);
        } catch (IOException e) {
            LogUtil.error(TAG, "Error touching file " + cacheFile + e.toString());
        }
    }

    /**
     * 注册缓存进度监听器
     *
     * @param cacheListener 缓存进度监听器
     * @param url           视频URL
     */
    public void registerCacheListener(CacheListener cacheListener, String url) {
        checkAllNotNull(cacheListener, url);
        synchronized (clientsLock) {
            try {
                getClients(url).registerCacheListener(cacheListener);
            } catch (ProxyCacheException e) {
                LogUtil.warn(TAG, "Error registering cache listener" + e.toString());
            }
        }
    }

    /**
     * 注销缓存进度监听器
     *
     * @param cacheListener 缓存进度监听器
     * @param url           视频URL
     */
    public void unregisterCacheListener(CacheListener cacheListener, String url) {
        checkAllNotNull(cacheListener, url);
        synchronized (clientsLock) {
            try {
                getClients(url).unregisterCacheListener(cacheListener);
            } catch (ProxyCacheException e) {
                LogUtil.warn(TAG, "Error registering cache listener" + e.toString());
            }
        }
    }

    /**
     * 注销缓存进度监听器
     *
     * @param cacheListener 缓存进度监听器
     */
    public void unregisterCacheListener(CacheListener cacheListener) {
        checkNotNull(cacheListener);
        synchronized (clientsLock) {
            for (HttpProxyCacheServerClients clients : clientsMap.values()) {
                clients.unregisterCacheListener(cacheListener);
            }
        }
    }

    /**
     * Checks is cache contains fully cached file for particular url.
     *
     * @param url an url cache file will be checked for.
     * @return {@code true} if cache contains fully cached file for passed in parameters url.
     */
    public boolean isCached(String url) {
        checkNotNull(url, "Url can't be null!");
        return getCacheFile(url).exists();
    }

    /**
     * 关闭代理客户端和服务端
     * 结束请求流程和缓存流程
     */
    public void shutdown() {
        LogUtil.info(TAG, "Shutdown proxy server");
        listener = null;
        shutdownClients();
        config.sourceInfoStorage.release();

        waitConnectionThread.interrupt();
        try {
            if (!serverSocket.isClosed()) {
                serverSocket.close();
            }
        } catch (IOException e) {
            onError(new ProxyCacheException("Error shutting down proxy server", e));
        }
    }

    private boolean isAlive() {
        return pinger.ping(3, 700);   // 70+140+280=max~500ms
    }

    private String appendToProxyUrl(String url) {
        try {
            return String.format(Locale.US, Constants.BASE_PROXY_URL, PROXY_HOST, port, ProxyCacheUtils.encode(url));
        } catch (IOException e) {
            return url;
        }
    }

    private File getCacheFile(String url) {
        File cacheDir = config.cacheRoot;
        String fileName = config.fileNameGenerator.generate(url);
        return new File(cacheDir, fileName);
    }

    private void touchFileSafely(File cacheFile) {
        try {
            config.diskUsage.touch(cacheFile);
        } catch (IOException e) {
            LogUtil.error(TAG, "Error touching file " + cacheFile + e.toString());
        }
    }

    /**
     * 关闭代理客户端
     */
    public void shutdownClients() {
        synchronized (clientsLock) {
            cacheOffset = 0;
            sameOffsetTimes = 0;
            for (HttpProxyCacheServerClients clients : clientsMap.values()) {
                clients.shutdown();
            }
            clientsMap.clear();
        }
    }


    private void processSocket(Socket socket) {
        try {
            GetRequest request = GetRequest.read(socket.getInputStream());
            LogUtil.debug(TAG, "Request to cache proxy: " + request);
            String url = ProxyCacheUtils.decode(request.uri);
            String newUrl = dealM3u8(url, originUrl);
            if (!pinger.isPingRequest(newUrl)) {
                if (cacheOffset == request.rangeOffset) {
                    sameOffsetTimes++;
                } else {
                    sameOffsetTimes = 0;
                }
                cacheOffset = request.rangeOffset;
                // 当同一个rangeOffset连续请求的时候可以判定为之前的请求失败，在疯狂重试
                // 这个时候需要作出处理(可能是分片/分节点策略存在问题 需要结束请求，切换为默认策略)
                if (sameOffsetTimes >= attemptsTime && listener != null) {
                    listener.reSetAfterError();
                    shutdown();
                    File cacheFile = getCacheFile(this.originUrl);
                    if (cacheFile != null) {
                        cacheFile.delete();
                    }
                    return;
                }
            }
            if (pinger.isPingRequest(newUrl)) {
                pinger.responseToPing(socket);
            } else {
                HttpProxyCacheServerClients clients = getClients(newUrl);
                clients.processRequest(request, socket);
            }
        } catch (SocketException e) {
            // There is no way to determine that client closed connection http://stackoverflow.com/a/10241044/999458
            // So just to prevent log flooding don't log stacktrace
            LogUtil.debug(TAG, "Closing socket… Socket is closed by client.");
        } catch (ProxyCacheException | InterruptedException | IOException e) {
            onError(new ProxyCacheException("Error processing request", e));
        } finally {
            releaseSocket(socket);
            LogUtil.debug(TAG, "Opened connections: " + getClientsCount());
        }
    }

    /**
     * 添加对m3u8格式视频的支持---可以播放 目前的缓存策略暂时无效
     *
     * @param url       socket解析出来的url eg:xxx01.ts
     * @param originUrl 初始请求的url  eg:http://devimages.apple.com.edgekey.net/streaming/examples/bipbop_4x3/gear4/prog_index.m3u8
     * @return 处理之后的url  eg:http://devimages.apple.com.edgekey.net/streaming/examples/bipbop_4x3/gear4/xxx01.ts
     */
    private String dealM3u8(String url, String originUrl) {
        if (TextTool.isNullOrEmpty(url) || TextTool.isNullOrEmpty(originUrl)) {
            return "";
        }
        if (!originUrl.toLowerCase().endsWith(".m3u8") || !originUrl.contains("/")) {
            return url;
        }
        if (url.toLowerCase().startsWith("http") || url.toLowerCase().startsWith("https")) {
            return url;
        }
        int index = originUrl.lastIndexOf("/") < 0 ? 0 : originUrl.lastIndexOf("/");
        String beginStr = originUrl.substring(0, index);
        return beginStr + "/" + url;
    }

    private HttpProxyCacheServerClients getClients(String url) throws ProxyCacheException {
        synchronized (clientsLock) {
            HttpProxyCacheServerClients clients = clientsMap.get(url);
            if (clients == null) {
                clients = new HttpProxyCacheServerClients(url, config, config.cacheStrategy);
                clientsMap.put(url, clients);
            }
            return clients;
        }
    }

    private int getClientsCount() {
        synchronized (clientsLock) {
            int count = 0;
            for (HttpProxyCacheServerClients clients : clientsMap.values()) {
                count += clients.getClientsCount();
            }
            return count;
        }
    }

    private void releaseSocket(Socket socket) {
        closeSocketInput(socket);
        closeSocketOutput(socket);
        closeSocket(socket);
    }

    private void closeSocketInput(Socket socket) {
        try {
            if (!socket.isInputShutdown() && !socket.isClosed()) {
                socket.shutdownInput();
            }
        } catch (SocketException e) {
            // There is no way to determine that client closed connection http://stackoverflow.com/a/10241044/999458
            // So just to prevent log flooding don't log stacktrace
            LogUtil.debug(TAG, "Releasing input stream… Socket is closed by client.");
        } catch (IOException e) {
            onError(new ProxyCacheException("Error closing socket input stream", e));
        }
    }

    private void closeSocketOutput(Socket socket) {
        try {
            if (!socket.isOutputShutdown() && !socket.isClosed()) {
                socket.shutdownOutput();
            }
        } catch (IOException e) {
            LogUtil.warn(TAG, "Failed to close socket on proxy side: {}. It seems client have already closed " +
                    "connection." + e.getMessage());
        }
    }

    private void closeSocket(Socket socket) {
        try {
            if (!socket.isClosed()) {
                socket.close();
            }
        } catch (IOException e) {
            onError(new ProxyCacheException("Error closing socket", e));
        }
    }

    private void onError(Throwable error) {
        LogUtil.error(TAG, "HttpProxyCacheServer error" + error.toString());
    }

    private final class WaitRequestsRunnable implements Runnable {
        private final CountDownLatch startSignal;
        private Socket oldSocket;

        WaitRequestsRunnable(CountDownLatch startSignal) {
            this.startSignal = startSignal;
        }

        @Override
        public void run() {
            startSignal.countDown();
            waitForRequest();
        }

        private void waitForRequest() {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    Socket socket = serverSocket.accept();
                    if (oldSocket != null) {
                        releaseSocket(oldSocket);
                    }
                    int localPort = socket.getLocalPort();
                    if (localPort != port) {
                        LogUtil.debug(TAG, "unknown socket attack,not local socket,port is : " + localPort);
                        continue;
                    }
                    LogUtil.debug(TAG, "Accept new socket {}" + socket);
                    oldSocket = socket;
                    socketProcessor.submit(new SocketProcessorRunnable(socket));
                }
            } catch (IOException e) {
                onError(new ProxyCacheException("Error during waiting connection", e));
            }
        }
    }

    private final class SocketProcessorRunnable implements Runnable {
        private final Socket socket;

        SocketProcessorRunnable(Socket socket) {
            this.socket = socket;
        }

        @Override
        public void run() {
            processSocket(socket);
        }
    }

    /**
     * Builder for {@link HttpProxyCacheServer}.
     */
    public static final class Builder {
        private static final long DEFAULT_MAX_SIZE = 512 * 1024 * 1024;

        private File cacheRoot;
        private FileNameGenerator fileNameGenerator;
        private DiskUsage diskUsage;
        private SourceInfoStorage sourceInfoStorage;
        private HeaderInjector headerInjector;
        private CacheStrategy cacheStrategy;


        public Builder(Context context) {
            this.sourceInfoStorage = SourceInfoStorageFactory.newSourceInfoStorage(context);
            this.cacheRoot = StorageUtils.getIndividualCacheDirectory(context);
            this.diskUsage = new TotalSizeLruDiskUsage(DEFAULT_MAX_SIZE);
            this.fileNameGenerator = new Md5FileNameGenerator();
            this.headerInjector = new EmptyHeadersInjector();
            this.cacheStrategy = new DefaultCacheStrategy();
        }

        /**
         * Overrides default cache folder to be used for caching files.
         * <p>
         * By default AndroidVideoCache uses
         * '/Android/data/[app_package_name]/cache/video-cache/' if card is mounted and app has appropriate permission
         * or 'video-cache' subdirectory in default application's cache directory otherwise.
         * </p>
         * <b>Note</b> directory must be used <b>only</b> for AndroidVideoCache files.
         *
         * @param file a cache directory, can't be null.
         * @return a builder.
         */
        public Builder cacheDirectory(File file) {
            this.cacheRoot = checkNotNull(file);
            return this;
        }

        /**
         * Overrides default cache file name generator {@link Md5FileNameGenerator} .
         *
         * @param fileNameGenerator a new file name generator.
         * @return a builder.
         */
        public Builder fileNameGenerator(FileNameGenerator fileNameGenerator) {
            this.fileNameGenerator = checkNotNull(fileNameGenerator);
            return this;
        }

        /**
         * Sets max cache size in bytes.
         * <p>
         * All files that exceeds limit will be deleted using LRU strategy.
         * Default value is 512 Mb.
         * </p>
         * Note this method overrides result of calling {@link #maxCacheFilesCount(int)}
         *
         * @param maxSize max cache size in bytes.
         * @return a builder.
         */
        public Builder maxCacheSize(long maxSize) {
            this.diskUsage = new TotalSizeLruDiskUsage(maxSize);
            return this;
        }

        /**
         * Sets max cache files count.
         * All files that exceeds limit will be deleted using LRU strategy.
         * Note this method overrides result of calling {@link #maxCacheSize(long)}
         *
         * @param count max cache files count.
         * @return a builder.
         */
        public Builder maxCacheFilesCount(int count) {
            this.diskUsage = new TotalCountLruDiskUsage(count);
            return this;
        }

        /**
         * Set custom DiskUsage logic for handling when to keep or clean cache.
         *
         * @param diskUsage a disk usage strategy, cant be {@code null}.
         * @return a builder.
         */
        public Builder diskUsage(DiskUsage diskUsage) {
            this.diskUsage = checkNotNull(diskUsage);
            return this;
        }

        /**
         * Add headers along the request to the server
         *
         * @param headerInjector to inject header base on url
         * @return a builder
         */
        public Builder headerInjector(HeaderInjector headerInjector) {
            this.headerInjector = checkNotNull(headerInjector);
            return this;
        }

        /**
         * Add CacheStrategy along the request to the server
         *
         * @param cacheStrategy the way of how to cache video
         * @return a builder
         */
        public Builder cacheStrategy(CacheStrategy cacheStrategy) {
            this.cacheStrategy = checkNotNull(cacheStrategy);
            return this;
        }

        /**
         * Builds new instance of {@link HttpProxyCacheServer}.
         *
         * @return proxy cache. Only single instance should be used across whole app.
         */
        public HttpProxyCacheServer build() {
            Config config = buildConfig();
            return new HttpProxyCacheServer(config);
        }

        private Config buildConfig() {
            return new Config(cacheRoot, fileNameGenerator, diskUsage, sourceInfoStorage, headerInjector,
                    cacheStrategy);
        }

    }
}
